#ifndef __SERVER_HPP__
#define __SERVER_HPP__

#include <iostream>
#include <vector>
#include <string>
#include <vector>
#include <unordered_map>
#include <algorithm>
#include <memory>
#include <thread>
#include <condition_variable>
#include <typeinfo>
#include <functional>
#include <cstdio>
#include <cstdint>
#include <cassert>
#include <cstring>
#include <ctime>
#include <cstdlib>
#include <cassert>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/timerfd.h>
#include <sys/eventfd.h>
#include <signal.h>

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG

#define LOG(level, format, ...)                                                             \
    do                                                                                      \
    {                                                                                       \
        if (level < LOG_LEVEL)                                                              \
            break;                                                                          \
        time_t t = time(nullptr);                                                           \
        struct tm *ltm = localtime(&t);                                                     \
        char tmp[32];                                                                       \
        memset(tmp, 0, sizeof(tmp));                                                        \
        strftime(tmp, 31, "%H:%M:%S", ltm);                                                 \
        fprintf(stdout, "[%s %s:%d] " format "\n", tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__);
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__);
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__);

class Buffer
{
private:
    static const int capacity = 1024;

private:
    char *Begin()
    {
        return &(*_buffer.begin());
    }

public:
    Buffer()
        : _buffer(capacity), _reader_idx(0), _writer_idx(0)
    {
    }

    // 获取当前写位置地址
    char *WriterPosition()
    {
        return Begin() + _writer_idx;
    }

    // 获取当前读位置地址
    char *ReaderPosition()
    {
        return Begin() + _reader_idx;
    }

    // 获取缓冲区末尾空闲空间大小
    uint64_t TailIdleSize()
    {
        return _buffer.size() - _writer_idx;
    }

    // 获取缓冲区起始空闲空间大小
    uint64_t HeadIdleSize()
    {
        return _reader_idx;
    }

    // 获取可读空间大小
    uint64_t ReadAbleSize()
    {
        return _writer_idx - _reader_idx;
    }

    // 将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        if (len == 0)
            return;
        assert(len <= ReadAbleSize());
        _reader_idx += len;
    }

    void MoveWriteOffset(uint64_t len)
    {
        if (len == 0)
            return;
        assert(len <= TailIdleSize());
        _writer_idx += len;
    }

    // 确保可写空间足够
    void EnsureWriteSpace(uint64_t len)
    {
        if (len <= TailIdleSize())
            return;
        if (len <= (TailIdleSize() + HeadIdleSize()))
        {
            // 移动数据
            uint64_t rsz = ReadAbleSize();
            std::copy(ReaderPosition(), ReaderPosition() + rsz, Begin());
            _reader_idx = 0;
            _writer_idx = rsz;
        }
        else
        {
            // 扩容
            _buffer.resize(_writer_idx + len);
        }
    }

    // 写入数据
    void Write(const void *data, uint64_t len)
    {
        if (len == 0)
            return;
        EnsureWriteSpace(len);
        const char *src = (const char *)data;
        std::copy(src, src + len, WriterPosition());
    }

    void WriteAndPush(const void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriteOffset(len);
    }

    void WriteString(const std::string &str)
    {
        Write(str.c_str(), str.size());
    }

    void WriteStringAndPush(const std::string &str)
    {
        WriteString(str);
        MoveWriteOffset(str.size());
    }

    void WriteBuffer(Buffer &data)
    {
        Write(data.ReaderPosition(), data.ReadAbleSize());
    }

    void WriteBufferAndPush(Buffer &data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }

    // 读取数据
    void Read(void *buff, uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::copy(ReaderPosition(), ReaderPosition() + len, (char *)buff);
    }

    void ReadAndPop(void *buff, uint64_t len)
    {
        Read(buff, len);
        MoveReadOffset(len);
    }

    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }

    std::string ReadAsStringAndPop(uint16_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }

    char *FindCRLF()
    {
        char *pc = (char *)memchr(ReaderPosition(), '\n', ReadAbleSize());
        return pc;
    }

    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == NULL)
        {
            return "";
        }
        return ReadAsString(pos - ReaderPosition() + 1);
    }

    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }

    void Clear()
    {
        _reader_idx = 0;
        _writer_idx = 0;
    }

    ~Buffer() = default;

private:
    std::vector<char> _buffer; // 缓冲区
    uint64_t _reader_idx;      // 读偏移
    uint64_t _writer_idx;      // 写偏移
};

class Sock
{
public:
    static const int g_backlog = 1024;

public:
    Sock()
        : _sockfd(-1)
    {
    }

    Sock(int sockfd)
        : _sockfd(sockfd)
    {
    }

    int Fd()
    {
        return _sockfd;
    }

    // 1.创建套接字
    bool Socket()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ERR_LOG("SERVER SOCKET FAILED!!");
            return false;
        }
        return true;
    }

    // 2.绑定
    bool Bind(const uint16_t &port, const std::string &ip = "0.0.0.0")
    {
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = inet_addr(ip.c_str());
        if (bind(_sockfd, (struct sockaddr *)&local, sizeof(local)) < 0)
        {
            ERR_LOG("SERVER BIND FAILED!!");
            return false;
        }
        return true;
    }

    // 3.监听
    bool Listen(int backlog = g_backlog)
    {
        if (listen(_sockfd, backlog) < 0)
        {
            ERR_LOG("SERVER LISTEN FAILED!!");
            return false;
        }
        return true;
    }

    // 4.获取新链接
    int Accept()
    {
        int sock = accept(_sockfd, nullptr, nullptr);
        if (sock < 0)
        {
            ERR_LOG("SERVER ACCEPT FAILED!!");
            return -1;
        }
        return sock;
    }

    // 5.链接
    bool Connect(const uint16_t &port, const std::string &ip)
    {
        struct sockaddr_in server;
        memset(&server, 0, sizeof(server));
        server.sin_family = AF_INET;
        server.sin_port = htons(port);
        server.sin_addr.s_addr = inet_addr(ip.c_str());
        if (connect(_sockfd, (struct sockaddr *)&server, sizeof(server)) < 0)
        {
            ERR_LOG("CLIENT CONNECT FAILED!!");
            return false;
        }
        return true;
    }

    // 6.读取数据
    ssize_t Recv(void *buff, size_t len, int flag = 0)
    {
        ssize_t s = recv(_sockfd, buff, len, flag);
        if (s <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ERR_LOG("SOCKET RECV FAILED!!");
            return -1;
        }
        return s;
    }

    // 6.1非阻塞读
    ssize_t NonBlockRecv(void *buff, size_t len)
    {
        return Recv(buff, len, MSG_DONTWAIT);
    }

    // 7.发送数据
    ssize_t Send(void *buff, size_t len, int flag = 0)
    {
        ssize_t s = send(_sockfd, buff, len, flag);
        if (s < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ERR_LOG("SOCKET SEND FAILED!!");
            return -1;
        }
        return s;
    }

    // 7.1非阻塞写
    ssize_t NonBlockSend(void *buff, size_t len)
    {
        if (len == 0)
            return 0;
        return Send(buff, len, MSG_DONTWAIT);
    }

    // 8.关闭套接字
    void Close()
    {
        if (_sockfd >= 0)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }

    // 9.设置套接字选项---开启地址端口复用
    void ReuseAddress()
    {
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
    }

    // 10.设置套接字属性---设置为非阻塞
    void SetNonBlock()
    {
        int fl = fcntl(_sockfd, F_GETFL);
        fcntl(_sockfd, F_SETFL, fl | O_NONBLOCK);
    }

    // 11.创建一个服务器链接
    bool CreateServer(const uint16_t &port, const bool &block_flag = false, const std::string &ip = "0.0.0.0")
    {
        // 1.创建套接字  2.设置是否阻塞属性 3.绑定套接字 4.监听套接字  5.设置地址端口复用
        if (!Socket())
            return false;
        if (block_flag)
            SetNonBlock();
        if (!Bind(port, ip))
            return false;
        if (!Listen())
            return false;
        ReuseAddress();
        return true;
    }

    // 12.创建一个客户端链接
    bool CreateClient(const uint16_t &port, const std::string &ip)
    {
        // 1.创建套接字  2.发起链接
        if (!Socket())
            return false;
        if (!Connect(port, ip))
            return false;

        DBG_LOG("client start success!!!");
        return true;
    }

    ~Sock()
    {
        Close();
    }

private:
    int _sockfd;
};

class EventLoop;

// Channel通过管理文件描述符从而设置各种事件回调函数
class Channel
{
private:
    typedef std::function<void()> EventCallback; // 事件回调函数

public:
    Channel(EventLoop *loop, int fd)
        : _fd(fd), _loop(loop), _events(0), _revents(0)
    {
    }

    // 获取当前文件描述符
    int Fd()
    {
        return _fd;
    }

    void SetREvents(uint32_t events)
    {
        _revents = events;
    } // 设置实际就绪的事件

    // 获取当前想要监控的事件
    uint32_t Events()
    {
        return _events;
    }

    // 设置可读事件回调
    void SetReadCallback(const EventCallback &cb)
    {
        _read_callback = cb;
    }

    // 设置可写事件回调
    void SetWriteCallback(const EventCallback &cb)
    {
        _write_callback = cb;
    }

    // 设置错误事件回调
    void SetErrorCallback(const EventCallback &cb)
    {
        _error_callback = cb;
    }

    // 设置关闭连接事件回调
    void SetCloseCallback(const EventCallback &cb)
    {
        _close_callback = cb;
    }

    // 设置任意事件回调
    void SetEventCallback(const EventCallback &cb)
    {
        _event_callback = cb;
    }

    // 当前是否监控了可读
    bool ReadAble()
    {
        return _events & EPOLLIN;
    }

    // 当前是否监控了可写
    bool WriteAble()
    {
        return _events & EPOLLOUT;
    }

    // 启动读事件监控
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    // 启动写事件监控
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 关闭读事件监控
    void DisableRead()
    {
        _events &= (~EPOLLIN);
        Update();
    }
    // 关闭写事件监控
    void DisableWrite()
    {
        _events &= (~EPOLLOUT);
        Update();
    }

    // 关闭所有事件监控
    void DisableAll()
    {
        _events = 0;
        Update();
    }

    void Remove();
    void Update();

    void HandleEvent()
    {
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if (_read_callback)
            {
                _read_callback();
            }
        }
        // 每一种可能会释放连接的处理操作，每次只处理一个
        if (_revents & EPOLLOUT)
        {
            if (_write_callback)
            {
                _write_callback();
            }
        }
        else if (_revents & EPOLLERR)
        {
            if (_error_callback)
            {
                _error_callback();
            }
        }
        else if (_revents & EPOLLHUP)
        {
            if (_close_callback)
            {
                _close_callback();
            }
        }

        if (_event_callback)
        {
            _event_callback();
        }
    }

private:
    int _fd;                       // 文件描述符
    EventLoop *_loop;              // 回指管理者
    uint32_t _events;              // 需要设置的事件
    uint32_t _revents;             // 当前链接触发的事件
    EventCallback _read_callback;  // 可读事件回调
    EventCallback _write_callback; // 可写事件回调
    EventCallback _error_callback; // 错误事件回调
    EventCallback _close_callback; // 连接断开事件被触发回调
    EventCallback _event_callback; // 任意事件回调
};

// Poller管理着所有的设置好了的Channel，并且记录着每一个加入到Epoll模型的Channel
// Poller中是对于所有监控事件的操作
class Poller
{
private:
    static const int max_epoll_events = 1024;

private:
    // 判断当前需要Channel是否已经在_channels管理中
    bool HasChannel(Channel *channel)
    {
        auto iter = _channels.find(channel->Fd());
        if (iter == _channels.end())
        {
            return false;
        }
        return true;
    }

    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event ev;
        memset(&ev, 0, sizeof(ev));
        ev.data.fd = fd;
        ev.events = channel->Events();
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if (ret < 0)
        {
            ERR_LOG("EPOLLCTL FAILED!!");
        }

        return;
    }

public:
    Poller()
        : _epfd(-1)
    {
        _epfd = epoll_create(max_epoll_events);
        DBG_LOG("epfd:%d", _epfd);
        if (_epfd < 0)
        {
            ERR_LOG("EPOLL CREATE FAILED!!");
            abort();
        }
    }

    // 对于添加或者修改Channel事件的监控
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == false)
        {
            // 不存在就添加进_channels模型中管理起来，并添加进epoll模型中
            int fd = channel->Fd();
            _channels.insert(std::make_pair(fd, channel));
            // 添加进epoll模型中
            return Update(channel, EPOLL_CTL_ADD);
        }

        return Update(channel, EPOLL_CTL_MOD);
    }

    // 将Channel监控从epoll模型和_channels模型中移除
    void RemoveEvent(Channel *channel)
    {
        auto iter = _channels.find(channel->Fd());
        if (iter != _channels.end())
        {
            _channels.erase(iter);
        }
        Update(channel, EPOLL_CTL_DEL);
    }

    // 开始监控,并且返回触发的活跃Channel
    void Poll(std::vector<Channel *> *active)
    {
        int nfds = epoll_wait(_epfd, _evs, max_epoll_events, -1); // 堵塞式在这里监控
        if (nfds < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("EPOLL WAIT FAILED:%s\n", strerror(errno));
            abort(); // 结束程序
        }

        for (int i = 0; i < nfds; i++)
        {
            int fd = _evs[i].data.fd;
            DBG_LOG("sock:%d", fd);
            auto iter = _channels.find(fd);
            assert(iter != _channels.end());
            iter->second->SetREvents(_evs[i].events); // 设置实际就绪的事件
            active->push_back(iter->second);
        }
        return;
    }

private:
    int _epfd;                                    // epoll模型fd
    struct epoll_event _evs[max_epoll_events];    // epoll数组
    std::unordered_map<int, Channel *> _channels; // 建立文件描述符和channel的映射关系从而管理其所有的Channel
};

// 定时器任务对象，通过设定定时器任务id来标识其唯一性
typedef std::function<void()> TaskFunc;
typedef std::function<void()> Release;
class TimerTask
{
private:
    uint64_t _id;      // 定时器任务ID
    uint32_t _timeout; // 定时任务超时时间
    bool _canceled;    // false--表示没有被取消，true--表示被取消
    TaskFunc _task_cb; // 当定时时间超时了，该去执行的任务
    Release _release;  // 通过回调的方式来将时间轮中存储在哈希表中的定时任务删除掉

public:
    TimerTask(uint64_t id, uint32_t delay, const TaskFunc &cb)
        : _id(id), _timeout(delay), _canceled(false), _task_cb(cb)
    {
    }

    void SetRelease(const Release &cb)
    {
        _release = cb;
    }

    void Cancel()
    {
        _canceled = true;
    }

    uint32_t DelayTime()
    {
        return _timeout;
    }

    ~TimerTask()
    {
        if (_canceled == false)
        {
            _task_cb();
        }
        _release();
    }
};

// 时间轮模块，管理着所有的定时任务，通过定时器任务id建立与定时器任务对象的映射关系，从而管理其所有的定时器任务对象
class TimerWheel
{
private:
    typedef std::shared_ptr<TimerTask> PtrTask;
    typedef std::weak_ptr<TimerTask> WeakTask;

private:
    int _tick;                                      // 滴答指针，当前秒针，走到哪里释放到哪里，就相当于是执行哪里的任务
    int _capacity;                                  // 记录时间轮长度，当前是秒级时间轮，默认是60
    std::vector<std::vector<PtrTask>> _wheel;       // 通过shared_ptr引用计数的原理来实现定时任务添加刷新，当引用计数为0的时候，析构调用函数回调执行定时任务
    std::unordered_map<uint64_t, WeakTask> _timers; // 建立定时任务ID和定时器任务对象的映射形成哈希表，从而管理所有的定时任务

    EventLoop *_loop; // 回指管理者
    int _timerfd;     // 定时器描述符--可读事件回调就是读取计数器，执行定时任务
    std::unique_ptr<Channel> _timer_channel;

private:
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ERR_LOG("TIMERFD CREATE FAILED!!");
            abort();
        }

        DBG_LOG("timerfd:%d", timerfd);
        // timerfd_settime(int fd,int flags,const struct itimerspec* new_value,struct itimerspec* old_value)
        struct itimerspec itime;
        memset(&itime, 0, sizeof(itime));
        // 设置超时时间为1秒后，每一次超时后，每次超时时间间隔为1秒
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &itime, nullptr);
        return timerfd;
    }

    int ReadTimerfd()
    {
        uint64_t times;
        int n = read(_timerfd, &times, sizeof(times));
        if (n < 0)
        {
            ERR_LOG("READ TIMEFD FAILED!!");
            abort();
        }
        return times;
    }

    // 这个函数每执行一次，就相当于时间轮秒针向后走一步，然后清理当前时间上的所有定时任务
    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();
    }

    void OnTime()
    {
        int times = ReadTimerfd();
        for (int i = 0; i < times; i++)
        {
            RunTimerTask();
        }
    }

    void RemoveTimer(uint64_t id)
    {
        auto iter = _timers.find(id);
        if (iter != _timers.end())
        {
            _timers.erase(iter);
        }
        return;
    }

    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb)
    {
        PtrTask pt(new TimerTask(id, delay, cb));
        pt->SetRelease(std::bind(&TimerWheel::RemoveTimer, this, id)); // 设置回调函数用来删除保存在哈希表中的映射关系
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(pt);
        _timers.insert(std::make_pair(id, WeakTask(pt)));
    }

    void TimerRefreshInLoop(uint64_t id)
    {
        auto iter = _timers.find(id);
        if (iter == _timers.end())
        {
            return; // 没有在_timers哈希表中保存，没有需要更新的定时任务对象，直接返回
        }

        PtrTask pt = iter->second.lock(); // lock获取weak_ptr管理的对象对应的shared_ptr
        int delay = pt->DelayTime();
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(pt);
    }

    void TimerCancelInLoop(uint64_t id)
    {
        auto iter = _timers.find(id);
        if (iter == _timers.end())
        {
            return; // 没找到定时任务，直接返回
        }

        PtrTask pt = iter->second.lock();
        if (pt)
        {
            pt->Cancel();
        }
    }

public:
    TimerWheel(EventLoop *loop)
        : _tick(0), _capacity(60), _wheel(_capacity), _loop(loop), _timerfd(CreateTimerfd()), _timer_channel(new Channel(loop, _timerfd))
    {
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this));
        _timer_channel->EnableRead(); // 启动读事件监控
    }

    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);

    // 这个接口存在线程安全问题，这个接口实际上不能被外界使用者使用，只能在模块内，在对应的EventLoop线程内执行
    bool HasTimer(uint64_t id)
    {
        auto iter = _timers.find(id);
        if (iter == _timers.end())
        {
            return false;
        }
        return true;
    }
};

// EventLoop模块管理着Poller模块，Channel模块，定时器TimerWheel模块
// EventLoop模块与线程一一对应
class EventLoop
{
private:
    typedef std::function<void()> Functor;

public:
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        DBG_LOG("_eventfd:%d", efd);
        if (efd < 0)
        {
            ERR_LOG("CREATE EVENTFD FAILED!!");
            abort();
        }
        return efd;
    }

    void ReadEventfd()
    {
        uint64_t res = 0;
        int n = read(_event_fd, &res, sizeof(res));
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            ERR_LOG("READ EVENTFD FAILED!!");
            abort();
        }
        return;
    }

    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _locK(_mutex);
            _tasks.swap(functor);
        }
        for (auto &func : functor)
        {
            func();
        }
        return;
    }

    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int n = write(_event_fd, &val, sizeof(val));
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            ERR_LOG("WRITE EVENTFD FAILED!!");
            return;
        }
        return;
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()), _event_fd(CreateEventFd()), _event_channel(new Channel(this, _event_fd)), _timer_wheel(this)
    {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        _event_channel->EnableRead();
    }

    void Start()
    {
        while (true)
        {
            // 1.进行事件监控
            std::vector<Channel *> active;
            _poller.Poll(&active);
            // 2.进行事件处理
            for (auto &channel : active)
            {
                channel->HandleEvent();
                // DBG_LOG("1");
            }
            // 3.执行任务池中的任务
            RunAllTask();
        }
    }

    // 用于判断当前线程是否是EventLoop对应的线程
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    void QueueInLoop(const Functor &cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }

        // 唤醒有可能因为没有事件就绪，而导致epoll阻塞
        // 其实就是给eventfd中写入一个数据，eventfd就会触发可读事件
        WeakUpEventFd();
    }

    // 判断当前要执行的任务是否处于当前线程中，如果是则执行，不是则压入队列
    void RunInLoop(const Functor &cb)
    {
        if (IsInLoop())
        {
            return cb();
        }

        return QueueInLoop(cb);
    }

    void UpdateEvent(Channel *channel)
    {
        return _poller.UpdateEvent(channel);
    }

    void RemoveEvent(Channel *channel)
    {
        return _poller.RemoveEvent(channel);
    }

    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
    {
        return _timer_wheel.TimerAdd(id, delay, cb);
    }

    void TimerRefresh(uint64_t id)
    {
        return _timer_wheel.TimerRefresh(id);
    }

    void TimerCancel(uint64_t id)
    {
        return _timer_wheel.TimerCancel(id);
    }

    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }

    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }

private:
    std::thread::id _thread_id; // EventLoop模块与线程一一对应，用_thread_id标识其绑定对应线程
    int _event_fd;              // eventfd唤醒IO事件监控有可能导致的阻塞
    std::unique_ptr<Channel> _event_channel;
    Poller _poller;              // 创建Poller对象，对所有文件描述符进行监控
    std::vector<Functor> _tasks; // 任务池
    std::mutex _mutex;           // 实现任务池操作的线程安全
    TimerWheel _timer_wheel;     // 定时器模块
};

// LoopThread模块，用于创建线程，并且在线程执行函数内部创建EventLoop，从而实现EventLoop和线程的一一对应的关系
class LoopThread
{
private:
    void ThreadEntry() // 线程入口函数
    {
        // 在线程执行函数中生成EventLoop
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }

public:
    LoopThread()
        : _loop(nullptr), _thread(std::thread(&LoopThread::ThreadEntry, this))
    {
    }

    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]()
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }

private:
    std::mutex _mutex;             // 互斥锁
    std::condition_variable _cond; // 条件变量
    EventLoop *_loop;              // 返回EventLoop指针，使得EventLoop与线程一一对应
    std::thread _thread;           // EventLoop对应的线程
};

// LoopThreadPool模块，对于线程池进行封装
class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop *base_loop)
        : _thread_count(0), _next_idx(0), _base_loop(base_loop)
    {
    }

    // 设置线程数
    void SetThreadCount(int thread_count)
    {
        _thread_count = thread_count;
    }

    // 创建线程池
    void Create()
    {
        if (_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for (int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
        return;
    }

    // 通过RR方式分配下一个线程
    EventLoop *NextLoop()
    {
        if (_thread_count == 0)
        {
            return _base_loop;
        }

        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }

private:
    int _thread_count;                  // 记录线程池中线程数
    int _next_idx;                      // 下一个该分配的线程
    EventLoop *_base_loop;              // 主Reactor
    std::vector<LoopThread *> _threads; // 线程池
    std::vector<EventLoop *> _loops;    // 从属Reactor
};

// 通用型Any
class Any
{
private:
    class holder
    {
    public:
        virtual ~holder(){};
        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };

    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val)
            : _val(val)
        {
        }

        // 获取子类对象保存的数据类型
        virtual const std::type_info &type()
        {
            return typeid(T);
        }

        // 针对当前对象自身，克隆出一个新的子类对象
        virtual holder *clone()
        {
            return new placeholder(_val);
        }

        T _val;
    };

    holder *_content;

public:
    Any()
        : _content(nullptr)
    {
    }

    template <class T>
    Any(const T &val)
        : _content(new placeholder<T>(val))
    {
    }

    Any(const Any &other)
        : _content(other._content ? other._content->clone() : nullptr)
    {
    }

    ~Any()
    {
        delete _content;
    }

    Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    // 返回子类对象保存的数据的指针
    template <class T>
    T *get()
    {
        assert(typeid(T) == _content->type());
        return &((placeholder<T> *)_content)->_val;
    }

    // 赋值运算符的重载函数
    template <class T>
    Any &operator=(const T &val)
    {
        // 为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原先保存的数据也就被释放
        Any(val).swap(*this);
        return *this;
    }
    Any &operator=(const Any &other)
    {
        Any(other).swap(*this);
        return *this;
    }
};

// 定义连接状态
typedef enum
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
} ConnStatu;

class Connection;
// Connection模块对于每一个链接进行封装管理，联合自定义的Socket模块、Buffer模块对于链接的操作进行整合管理
typedef std::shared_ptr<Connection> PtrConnection;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    typedef std::function<void(const PtrConnection &)> ConnectedCallback;
    typedef std::function<void(const PtrConnection &, Buffer *)> MessageCallback;
    typedef std::function<void(const PtrConnection &)> ClosedCallback;
    typedef std::function<void(const PtrConnection &)> AnyEventCallback;

private:
    uint64_t _conn_id;             // 连接的唯一ID，便于连接的管理和查找
    int _sockfd;                   // 连接关联的文件描述符
    bool _enable_inactive_release; // 连接是否启动非活跃销毁的判断标志，默认是false
    EventLoop *_loop;              // 回指管理者,连接所关联的一个EventLoop
    ConnStatu _statu;              // 连接状态
    Sock _sock;                    // 套接字管理
    Channel _channel;              // 连接事件管理
    Buffer _in_buffer;             // 输入缓冲区
    Buffer _out_buffer;            // 输出缓冲区
    Any _context;                  // 请求的接收处理上下文

    // 这四个回调函数，是让服务器模块来设置的（其实服务器模块的处理回调也是组件使用者设置的）
    // 换句话说，这几个回调都是组件使用者设置的
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

    // 组件内连接关闭回调---组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭
    // 就应该从管理的地方移除掉自己的信息
    ClosedCallback _server_closed_callback;

private:
    void HandleRead()
    {
        // 1.接收socket的数据，放到缓冲区中
        char buffer[65536];
        ssize_t ret = _sock.NonBlockRecv(buffer, sizeof(buffer));
        if (ret < 0)
        {
            return ShutdownInLoop();
        }
        _in_buffer.WriteAndPush(buffer, ret);
        if (_in_buffer.ReadAbleSize() > 0)
        {
            return _message_callback(shared_from_this(), &_in_buffer);
        }
    }

    void HandleWrite()
    {
        ssize_t ret = _sock.NonBlockSend(_out_buffer.ReaderPosition(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            if (_in_buffer.ReadAbleSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            return Release(); // 这个时候才是实际的关闭操作
        }

        _out_buffer.MoveReadOffset(ret);
        if (_out_buffer.ReadAbleSize() == 0)
        {
            _channel.DisableWrite();
            // 如果当前是连接待关闭状态，则有数据，发送完数据释放连接，没有数据则直接释放
            if (_statu == DISCONNECTING)
            {
                return Release();
            }
        }
        return;
    }

    void HandleClose()
    {
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        return Release();
    }

    void HandleError()
    {
        return HandleClose();
    }

    void HandleEvent()
    {
        if (_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id);
        }
        if (_event_callback)
        {
            _event_callback(shared_from_this());
        }
    }

    void EstablishedInLoop()
    {
        assert(_statu == CONNECTING);
        _statu = CONNECTED;
        _channel.EnableRead();
        if (_connected_callback)
        {
            _connected_callback(shared_from_this());
        }
    }

    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimer(_conn_id))
        {
            _loop->TimerCancel(_conn_id);
        }
    }

    // 这个才是真正的释放接口
    void ReleaseInLoop()
    {
        _statu == DISCONNECTED;
        _channel.Remove();
        _sock.Close();
        if (_loop->HasTimer(_conn_id))
        {
            CancelInactiveReleaseInLoop();
        }
        if (_closed_callback)
        {
            _closed_callback(shared_from_this());
        }
        if (_server_closed_callback)
        {
            _server_closed_callback(shared_from_this());
        }
    }

    // 这个接口并不是实际发送接口，只是将数据放到了发送缓冲区，启动了可写事件监控
    void SendInLoop(Buffer &buf)
    {
        if (_statu == DISCONNECTED)
        {
            return;
        }
        _out_buffer.WriteBufferAndPush(buf);
        if (_channel.WriteAble() == false)
        {
            _channel.EnableWrite();
        }
    }

    void ShutdownInLoop()
    {
        _statu = DISCONNECTING; // 设置连接为半关闭状态
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        // 要么就是写入数据的时候出错关闭，要么就是没有待发送数据，直接关闭
        if (_out_buffer.ReadAbleSize() > 0)
        {
            if (_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        if (_out_buffer.ReadAbleSize() == 0)
        {
            Release();
        }
    }

    void EnableInactiveReleaseInLoop(int sec)
    {
        // 1. 将判断标志 _enable_inactive_release 置为true
        _enable_inactive_release = true;
        // 2. 如果当前定时销毁任务已经存在，那就刷新延迟一下即可
        if (_loop->HasTimer(_conn_id))
        {
            return _loop->TimerRefresh(_conn_id);
        }
        // 3. 如果不存在定时销毁任务，则新增
        _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
    }

    void UpgradeInLoop(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        : _conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false), _loop(loop), _statu(CONNECTING), _sock(_sockfd), _channel(_loop, _sockfd)
    {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    }

    ~Connection()
    {
        DBG_LOG("RELEASE CONNECTION:%p", this);
    }

    // 获取管理的文件描述符
    int Fd()
    {
        return _sockfd;
    }

    // 获取连接ID
    int Id()
    {
        return _conn_id;
    }

    bool Connected()
    {
        return (_statu == CONNECTED);
    }

    void SetContext(const Any &context)
    {
        _context = context;
    }

    Any *GetContext()
    {
        return &_context;
    }

    // 设置回调指针(给上层提供的接口，由上层来指定干嘛)
    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _closed_callback = cb;
    }
    void SetSrvClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }

    // 连接建立就绪后，进行channel回调设置，启动读监控，调用_connected_callback
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }

    // 发送数据，将数据放到发送缓冲区，启动写事件监控
    void Send(const char *data, size_t len)
    {
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }

    // 提供给组件使用者的关闭接口---并不实际关闭，需要判断有没有数据待处理
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }

    void Release()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }

    // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }

    // 取消非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }

    void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
};

// Acceptor模块，对于监听套接字进行封装
class Acceptor
{
private:
    typedef std::function<void(int)> AcceptCallback;

private:
    int CreateServer(uint16_t port)
    {
        bool ret = _sock.CreateServer(port);
        assert(ret == true);
        return _sock.Fd();
    }

private:
    void HandleRead()
    {
        int newfd = _sock.Accept();
        if (newfd < 0)
        {
            return;
        }

        DBG_LOG("accept new link");
        if (_accept_callback)
            _accept_callback(newfd);
    }

public:
    // 不再构造函数内部启动读事件监控，因为可能此时_accept_callback还没被设置，一旦有新连接到来，就会造成资源泄漏
    Acceptor(EventLoop *loop, uint16_t port)
        : _sock(CreateServer(port)), _loop(loop), _channel(_loop, _sock.Fd())
    {
        DBG_LOG("acceptorFd:%d", _sock.Fd());
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }

    // 启动读事件监控
    void Listen()
    {
        _channel.EnableRead();
    }

    // 设置回调
    void SetAcceptCallback(const AcceptCallback &cb)
    {
        _accept_callback = cb;
    }

private:
    Sock _sock;       // 套接字
    EventLoop *_loop; // 主Reactor对应
    Channel _channel; // 监控事件管理
    AcceptCallback _accept_callback;
};

class TcpServer
{
private:
    typedef std::function<void(const PtrConnection &)> ConnectedCallback;
    typedef std::function<void(const PtrConnection &, Buffer *)> MessageCallback;
    typedef std::function<void(const PtrConnection &)> ClosedCallback;
    typedef std::function<void(const PtrConnection &)> AnyEventCallback;
    typedef std::function<void()> Functor;

private:
    uint64_t _next_id;                                  // 一个自增长的连续ID
    uint16_t _port;                                     // 端口
    int _timeout;                                       // 超时时间间隔
    bool _enable_inactive_release;                      // 是否开启非活跃连接超时销毁的判断标志
    EventLoop _baseloop;                                // 主Reactor
    Acceptor _acceptor;                                 // 对于监听套接字进行封装(服务器套接字的前期所有操作都封装在这里面)
    LoopThreadPool _pool;                               // 线程池
    std::unordered_map<uint64_t, PtrConnection> _conns; // 保存管理所有连接对应的shared_ptr对象

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

private:
    void NewConnection(int fd)
    {
        _next_id++;
        PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if (_enable_inactive_release)
            conn->EnableInactiveRelease(_timeout); // 启动非活跃超时销毁
        conn->Established();                       // 就绪初始化
        _conns.insert(std::make_pair(_next_id, conn));
        DBG_LOG("1");
    }

    void RunAfterInLoop(const Functor &task, int delay)
    {
        _next_id++;
        _baseloop.TimerAdd(_next_id, delay, task);
    }

    void RemoveConnectionInLoop(const PtrConnection &conn)
    {
        int id = conn->Id();
        auto it = _conns.find(id);
        if (it != _conns.end())
        {
            _conns.erase(it);
        }
    }

    // 从管理Connection的_conns中移除连接信息
    void RemoveConnection(const PtrConnection &conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }
    //_acceptor在初始化列表阶段完成sock的准备工作
public:
    TcpServer(uint16_t port)
        : _next_id(0), _port(port), _enable_inactive_release(false), _acceptor(&_baseloop, _port), _pool(&_baseloop)
    {
        // 为其设定NewConnection回调
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        // 而后为其开启读事件监控
        _acceptor.Listen();
    }

    // 设置线程数
    void SetThreadCount(int count)
    {
        _pool.SetThreadCount(count);
    }

    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }

    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }

    void SetClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }

    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _event_callback = cb;
    }

    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }

    // 添加一个定时任务
    void RunAfter(const Functor &task, int delay)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }

    void Start()
    {
        _pool.Create();
        _baseloop.Start();
    }
};

class NetWork
{
public:
    NetWork()
    {
        DBG_LOG("SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};

void Channel::Update()
{
    return _loop->UpdateEvent(this);
}

void Channel::Remove()
{
    return _loop->RemoveEvent(this);
}

void TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
{
    return _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb));
}

void TimerWheel::TimerRefresh(uint64_t id)
{
    return _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
}

void TimerWheel::TimerCancel(uint64_t id)
{
    return _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}

#endif